package site.edody.dframe.rocketmq;

import org.springframework.beans.factory.annotation.Autowired;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.ssssssss.magicapi.core.service.MagicAPIService;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class ConsumerService {
    private final Logger logger = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    MagicAPIService service;

    public ConsumeConcurrentlyStatus processMessages(List<MessageExt> msgs, String topic) {
        for (MessageExt msg : msgs) {
            try{
                logger.debug("\n======= 收到消息 =======");
                logger.debug("主题: " + topic);
                logger.debug("标签: " + msg.getTags());
                logger.debug("内容: " + new String(msg.getBody()));
                logger.debug("消息ID: " + msg.getMsgId());
                Map<String, Object> params = new HashMap<>();
                // 注入变量信息
                params.put("content", new String(msg.getBody()));
                params.put("id", msg.getMsgId());
                params.put("tags",msg.getTags());
                params.put("topic",topic);
                // 调用函数
                Object value = service.invoke("/mq/onmessage", params);
                logger.info("调用onMessage结束："+value.toString());
                logger.debug("========================\n");
            }catch (Exception e){
                logger.error("Process MQ Message Exception",e);
            }


        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}